package org.example.graphx.basic

import org.apache.spark.graphx.{Edge, Graph}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.slf4j.LoggerFactory

object ConstructGraph {

  def main(args: Array[String]): Unit = {

    var log = LoggerFactory.getLogger(this.getClass.getName)

    //设置运行环境
    val conf = new SparkConf()
      .setAppName("graphx-constructGraph")
      .setMaster("local[1]")
    val sc = new SparkContext(conf)

    //设置顶点和边，注意顶点和边都是用元组定义的Array
    //顶点的数据类型是VD:(String,Int)
    val vertexArray = Array(
      (1L, ("Alice")),
      (2L, ("Bob")),
      (3L, ("Charlie")),
      (4L, ("David")),
      (5L, ("Ed")),
      (6L, ("Fran"))
    )

    //边的数据类型ED:Int
    val edgeArray = Array(
      Edge(2L, 1L, 1),
      Edge(2L, 4L, 1),
      Edge(3L, 2L, 1),
      Edge(3L, 6L, 1),
      Edge(4L, 1L, 1),
      Edge(5L, 2L, 1),
      Edge(5L, 3L, 1),
      Edge(5L, 6L, 1)
    )

    //构造vertexRDD和edgeRDD
    val vertexRDD: RDD[(Long, (String))] = sc.parallelize(vertexArray)
    val edgeRDD: RDD[Edge[Int]] = sc.parallelize(edgeArray)

    //构造图Graph[VD,ED]
    val graph: Graph[(String), Int] = Graph(vertexRDD, edgeRDD)

    log.warn("page rank result：")
    val ranks = graph.pageRank(0.0001).vertices
    println(ranks.sortBy(_._2, false).collect().take(3).mkString("\n"))

    log.warn("connected components result：")
    val cc = graph.connectedComponents().vertices
    println(cc.collect().mkString("\n"))

    log.warn("triangle count result：")
    val triCounts = graph.triangleCount().vertices
    println(triCounts.collect().mkString("\n"))

    sc.stop()
  }

}
